package io.mqttpush.mqttclient.conn;

import io.mqttpush.mqttclient.conn.UnLoginTask.TopicMsg;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 
 * @author tianzhenjiu
 *
 */
public class DefaultOptionsProvider {

	Channel channel;
	
	Logger logger= LoggerFactory.getLogger(getClass());
	
	
	static DefaultOptionsProvider defaultOptionsProvider;
	
	
	
	private DefaultOptionsProvider() {
		
	}
	
	/**
	 * 单例模式
	 * @return
	 */
	public  static DefaultOptionsProvider instance(){
		
		if(defaultOptionsProvider==null){
			defaultOptionsProvider=new DefaultOptionsProvider();
		}
		return defaultOptionsProvider;
	}
	
	
	public ChannelFuture login(String deviceId,String username, String password) {
		
		MqttFixedHeader mqttFixedHeader=new MqttFixedHeader(MqttMessageType.CONNECT,
				false, MqttQoS.AT_MOST_ONCE, false, 0);
		
		MqttConnectVariableHeader variableHeader=new MqttConnectVariableHeader(
				MqttVersion.MQTT_3_1_1.protocolName(), MqttVersion.MQTT_3_1_1.protocolLevel(), true, true, false, 0,false, false, 10);
		
		MqttConnectPayload payload=new MqttConnectPayload(deviceId, null, null, username, password.getBytes());
		MqttConnectMessage connectMessage=new MqttConnectMessage(mqttFixedHeader, variableHeader, payload);

		return channel.writeAndFlush(connectMessage);
	}

	
	public Future<?> subscribe(String topname, MqttQoS qoS) {



		if(!islogin()){
			
			RuntimeException cause=null;
			
			if(channel==null) {
				return new FailFuture(new RuntimeException("channel is null"));
			}
			
			return channel.newFailedFuture(new RuntimeException("need login first"));
			
		}
		Integer msgId=Math.abs(Integer.valueOf(topname.hashCode()).shortValue());

		List<MqttTopicSubscription> topicSubscriptions=new ArrayList<>();
		topicSubscriptions.add(new MqttTopicSubscription(topname, qoS));
		
		
		MqttFixedHeader mqttFixedHeader=
				new MqttFixedHeader(MqttMessageType.SUBSCRIBE,false, MqttQoS.AT_LEAST_ONCE,false , 0);
		
		MqttMessageIdVariableHeader variableHeader=
				MqttMessageIdVariableHeader.from(msgId);
		MqttSubscribePayload payload=new MqttSubscribePayload(topicSubscriptions);
		MqttSubscribeMessage mqttSubscribeMessage=new MqttSubscribeMessage(mqttFixedHeader, variableHeader, payload);

		logger.info("开始订阅{}->{}",topname,String.valueOf(msgId));
		return getChannel().writeAndFlush(mqttSubscribeMessage);
	}

	
	public Future<?> unsub(String topname) {

		if(!islogin()){
			
			
			RuntimeException cause=null;;
			
			if(!UnLoginTask.untopics.offer(topname)) {
				cause=new RuntimeException("待发队列满了");
			}
			else {
				cause=new RuntimeException("未登录");
			}
			
			if(channel==null) {
				return new FailFuture(cause);
			}
			
			return channel.newFailedFuture(cause);
			
		
		}
		
		
		List<String> topicSubscriptions=new ArrayList<>();
		topicSubscriptions.add(topname);

		Integer msgId=Math.abs(Integer.valueOf(topname.hashCode()).shortValue());
		
		MqttFixedHeader mqttFixedHeader=
				new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE,false, MqttQoS.AT_LEAST_ONCE,false , 0);
		
		MqttMessageIdVariableHeader variableHeader=
				MqttMessageIdVariableHeader.from(msgId);
		MqttUnsubscribePayload payload=new MqttUnsubscribePayload(topicSubscriptions);
		MqttUnsubscribeMessage mqttSubscribeMessage=new MqttUnsubscribeMessage(mqttFixedHeader, variableHeader, payload);

		logger.info("取消订阅{}->{}",topname,String.valueOf(msgId));
		return getChannel().writeAndFlush(mqttSubscribeMessage);
	}

	
	public Future<?> pubMsg(String topname, byte[] bs, MqttQoS qoS) {
		
		if(!islogin()){
			
			RuntimeException cause=null;
			
			if(!UnLoginTask.unLoginMsg.offer(new TopicMsg(topname, qoS, bs))) {
				cause=new RuntimeException("待发队列满了");
			}
			else {
				cause=new RuntimeException("未登录");
			}
			
			if(channel==null) {
				return new FailFuture(cause);
			}
			
			return channel.newFailedFuture(cause);
		
		}
		
			
		ByteBuf byteBuf=Unpooled.wrappedBuffer(bs);
		
		int messageid=Math.abs(Integer.valueOf(bs.hashCode()).shortValue());
		MqttFixedHeader mqttFixedHeader=
				new MqttFixedHeader(MqttMessageType.PUBLISH,false, qoS,false , 0);
 		
 		MqttPublishVariableHeader variableHeader=
 				new MqttPublishVariableHeader(topname,messageid);
 		
 		MqttPublishMessage mqttPublishMessage=
 				new MqttPublishMessage(mqttFixedHeader, variableHeader, byteBuf);
 		return getChannel().writeAndFlush(mqttPublishMessage);
	}

	public Channel getChannel() {
		return channel;
	}

	public void setChannel(Channel channel) {
		this.channel = channel;
	}

	/**
	 * 判断是否登录
	 * @return
	 */
	boolean islogin(){
		AtomicBoolean isLogin=Status.isLogin;
		
		return isLogin!=null&&isLogin.get();
		
	}
}
